package com.asiainfo.dacp.datastash.broker.flow.listener.service;

import java.util.List;
import java.util.Map;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Service;

import com.asiainfo.dacp.datastash.broker.core.BrokerException;
import com.asiainfo.dacp.datastash.broker.core.enums.UnitSchedType;
import com.asiainfo.dacp.datastash.broker.core.enums.UnitStepTypeEnum;
import com.asiainfo.dacp.datastash.broker.flow.core.domain.Flow;
import com.asiainfo.dacp.datastash.broker.flow.core.service.FlowCompileService;
import com.asiainfo.dacp.datastash.broker.flow.core.service.FlowStateChangeService;
import com.asiainfo.dacp.datastash.broker.flow.depend.service.DependJudger;
import com.asiainfo.dacp.datastash.broker.flow.depend.service.FileDependJudger;
import com.asiainfo.dacp.datastash.broker.flow.depend.service.SqlDependJudger;
import com.asiainfo.dacp.datastash.broker.redis.service.RedisFlowService;
import com.asiainfo.dacp.datastash.domain.unit.models.DataStashUnit;
import com.asiainfo.dacp.datastash.domain.unit.models.DataStashUnitStep;
import com.google.common.base.Strings;

@Service
public class UnitWaitProcessor {

	private static Logger LOG = LoggerFactory.getLogger(UnitWaitProcessor.class);
	@Autowired
	private RedisFlowService redisService;

	@Autowired
	private FlowStateChangeService stateService;

	@Autowired
	private FlowCompileService compileService;

	@Async
	protected void handle() {

		String serNo = redisService.fetchWaitUnit();
		if (Strings.isNullOrEmpty(serNo))
			return;

		Flow flow = redisService.getFlow(serNo);
		LOG.info(String.format("[%s] 开始进行依赖判断.", flow.buildSerNo()));
		if (flow != null) {
			boolean dependJudge = false;
			try {
				dependJudge = dependJudge(flow);
			} catch (BrokerException e) {
				e.printStackTrace();
				LOG.error(String.format("[%s] 依赖判断过程失败，不再入队.", flow.buildSerNo()), e);
				return;
			}
			if (dependJudge) {
				LOG.info(String.format("[%s] 依赖判断通过.", flow.buildSerNo()));
				stateService.dependOK(flow);
				try {
					compileService.compile(flow);
					stateService.compileOK(flow);
				} catch (Exception e) {
					stateService.compileNotOK(flow);
					LOG.error(String.format("[%s] 编译过程失败，不再入队.", flow.buildSerNo()), e);
					return;
				}
				redisService.putTasksWait(flow);
			} else {
				LOG.info(String.format("[%s] 依赖判断未通过，重新入队.", flow.buildSerNo()));
				stateService.dependNotOK(flow);
			}
			redisService.createOrUpdate(flow);
		}
	}

	private boolean dependJudge(Flow flow) throws BrokerException {
		DependJudger judger = getDependJudger(flow.getUnit());
		if (judger == null) {
			return true;
		}
		return judger.judge(flow);
	}

	private DependJudger getDependJudger(DataStashUnit unit) {
		List<DataStashUnitStep> steps = unit.getUnitSteps();
		for (DataStashUnitStep step : steps) {
			if (UnitStepTypeEnum.Schedule.value().equals(step.getStepDefine().getStepType())) {
				Map<String, Object> stepCfg = step.getStepCfgValMap();
				String schedType = (String) stepCfg.get("schedType");
				if (UnitSchedType.File.value().equals(schedType)) {
					return new FileDependJudger(stepCfg);
				}
				if (UnitSchedType.SQL.value().equals(schedType)) {
					return new SqlDependJudger(stepCfg);
				}
				break;
			}
		}
		return null;
	}

}
